/* Include files. */
extern "C" { 
	#include "standard.h"
	#include "util.h"
	#include "debug.h"
}
#include "timercore.h"
#include "child.h"
#include "script.h"
#include "cplusplus.h"

/* Macro constant definitions. */

/* Local function declarations. */

/* Macro API definitions. */

/* Global variable declarations. */

int parent_pipe[2];

#ifdef U_LOOP
static uloop_fd child_handle[2] = {0};
#else
static struct ev_loop *child_loop;
static ev_io child_handle[2];
#endif

#define INITSTATUS 0xFFFF

struct pid_status{
	int pid;
};

static queue<struct script*> script_result_queue;
static map<event*, struct script*> script_result_map;
static map<event*, struct pid_status> pid_map;

/* Type definitions. */

/* Local function definitions. */

int pipeRead(int fd, char *buf, int len, int *finish)
{
	int readed = 0;
	int error = 0;
	int ret;

	do {
		ret = recv(fd, &buf[readed], len - readed, 0);
		if(ret > 0) {
			readed += ret;
		} else if (0 == ret) {
			error = 1;
			break;
		} else {
			if (TCP_NORMAL_ERR(errno)) {
				break;
			} else {
				error = 1;
				break;
			}
		}
	} while(readed < len);

	if (finish != NULL)
		*finish = error;
	return readed;
}

int pipeWrite(int fd, const char *buf, int len)
{
	int writed = 0;
	int ret;

	do {
		ret = send(fd, &buf[writed], len - writed, MSG_NOSIGNAL);
		if(ret > 0) {
			writed += ret;
		} else if(0 == ret) {
			break;
		} else {
			if(TCP_NORMAL_ERR(errno)) {
				break;
			} else {
				writed = -1;
				break;
			}
		}
	} while(writed < len);

	return writed;
}

static int readHeader(int fd, struct script *s)
{	
	int ret = pipeRead(fd, (char *)&(*s), sizeof(*s));
	if (ret > 0){
		ASSERT(ret == sizeof(*s));
	}
	return ret;
}

#ifdef U_LOOP
static void signalHandler(struct uloop_process *c, int status)
{
	event* ev = (event*)c->data;
	int code ;
	map<event*, struct pid_status>::iterator ite = pid_map.find(ev); 
	if(ite != pid_map.end()){
		struct pid_status p = pid_map[ev];
		if (p.pid == c->pid) {
			if (WIFEXITED(status)) {
		    	code = WEXITSTATUS(status); 
		    }    
		    else {
				WARN("exec shell error, status = %d\n", status);
		        code = -1;  
		    }
			struct script *r = (struct script *)malloc(sizeof(*r));
			memset(r, 0, sizeof(*r));
			r->pid = p.pid;
			r->ev = ev;
			r->len = 0;
			r->type = e_SYNC;
			r->finish = e_SHELL_FINISH;
			r->code = code;
			pid_map.erase(ev);			
			script_result_queue.push(r);
#ifdef U_LOOP
			uloop_fd_add(&child_handle[0], ULOOP_WRITE | ULOOP_READ);
#else
			ev_io_stop(loop, &child_handle[1]);
			ev_io_init(&child_handle[1], sendResult, parent_pipe[0], EV_WRITE);
			ev_io_start(loop, &child_handle[1]);
#endif
			free(c);
		}
	}
}
#endif

#ifdef U_LOOP
static void sendResult(uloop_fd *h, unsigned int revents)
#else
static void sendResult(struct ev_loop *loop, ev_io *h, int revents)
#endif
{
	while(script_result_queue.size()){
		struct script *r = script_result_queue.front();
		script_result_queue.pop();
		int total = sizeof(*r) + r->len;
		int ret = pipeWrite(h->fd, (const char *)r, total);
		if (ret < 0) {
			script_result_queue.push(r);
			return;
		}
		else if (ret == 0){
			script_result_queue.push(r);
			return;
		}
		else {
			ASSERT(ret == total);
			free(r);
		}
	}
	if (!script_result_queue.size())
#ifdef U_LOOP
		uloop_fd_add(h, h->flags & (~ULOOP_WRITE));
#else
		ev_io_stop (loop, h);
#endif

	return;
}

#ifdef U_LOOP
static void readScriptResult(uloop_fd *h, unsigned int revents)
#else
static void readScriptResult(struct ev_loop *loop, ev_io *h, int revents)
#endif
{
	struct pid_status p;
	char buff[MAX_SCRIPT_RESULT_LEN];
	int finish;
	int ret = 0;

	ret = pipeRead(h->fd, (char *)buff, MAX_SCRIPT_RESULT_LEN, &finish);
	event* ev = (event*)h->data;
	map<event*, struct pid_status>::iterator ite = pid_map.find(ev); 
	if(ite != pid_map.end()){
		p = pid_map[ev];
		struct script *r = (struct script *)malloc(sizeof(*r) + ret);
		memset(r, 0, sizeof(*r) + ret);
		r->pid = p.pid;
		r->ev = ev;
		r->len = ret;
		r->type = e_SYNC;
		if (r->len > MAX_SCRIPT_RESULT_LEN) {
			ASSERT(0);
		}
		if (ret > 0) {
			memcpy(r->data, buff, r->len);
		}
		if (finish) {
			r->finish = e_SHELL_MAX;
		}
		script_result_queue.push(r);
#ifdef U_LOOP
		uloop_fd_add(&child_handle[0], ULOOP_WRITE | ULOOP_READ);
#else
		ev_io_stop(loop, &child_handle[1]);
		ev_io_init(&child_handle[1], sendResult, parent_pipe[0], EV_WRITE);
		ev_io_start(loop, &child_handle[1]);
#endif
	}
	if (finish) {
#ifdef U_LOOP
		uloop_fd_delete(h);
#else
		ev_io_stop(loop, h);
#endif
		close(h->fd);
		free(h);
	}
}

static int forkRunScript(const char *argv[], struct script *s)
{
	pid_t pid;
	int pipe[2];

	if(socketpair(AF_UNIX, SOCK_STREAM, 0, pipe) == -1) {
		ERROR("socketpair: %s\n", strerror(errno));
		return 0;
	} 
	pid = fork();	 
	switch (pid) {
		case -1:
			ERROR("fork %s: %s\n", argv[0], strerror(errno));
			return 0;
		case 0:
			/* child */
			close(pipe[1]);
			switch(s->type) {
				case e_ASYNC:
					close(1);
					close(2);
					break;
				case e_SYNC:
					dup2(pipe[0], STDOUT_FILENO); 
					dup2(pipe[0], STDERR_FILENO); 
					break;
			}
			execvp(argv[0], (char * const *)argv);
			ERROR("pid = %d over\n", pid);
			exit(0);
		default:
			/* parent */
			close(pipe[0]); 
			break;
	}

	switch(s->type) {
		case e_ASYNC:
			close(pipe[1]);
			break;
		case e_SYNC:
			int flags = fcntl(pipe[1], F_GETFL);
			fcntl(pipe[1], F_SETFL, flags | O_NONBLOCK);		
#ifdef U_LOOP
			struct uloop_fd *h = (struct uloop_fd *)malloc(sizeof(*h));
			memset(h, 0, sizeof(*h));
			h->data = s->ev;
			h->fd = pipe[1];
			h->cb = readScriptResult;
			uloop_fd_add(h, ULOOP_READ);
			
			struct uloop_process *p = (struct uloop_process *)malloc(sizeof(*p));
			memset(p, 0, sizeof(*p));
			p->cb = signalHandler;
			p->pid = pid;
			p->data = s->ev;
			uloop_process_add(p);
#else
			ev_io *h = (ev_io *)malloc(sizeof(*h));
			h->data = s->ev;
			ev_io_init (h, readScriptResult, pipe[1], EV_READ);
			ev_io_start (child_loop, h);
#endif
			break;
	}
	
	return pid;
}

#ifdef U_LOOP
static void readScript(struct uloop_fd *h, unsigned int revents)
#else
static void readScript(struct ev_loop *loop, ev_io *h, int revents)
#endif
{
	struct script s;
	int pid;
	char buff[MAX_SCRIPT_LEN];
	map<event*, struct pid_status>::iterator ite;
	struct pid_status p;

#ifdef U_LOOP
	if(revents & ULOOP_WRITE) {
		sendResult(h, revents);
	}

	if(!(revents & ULOOP_READ)) {
		return;
	}
#endif
	
	int ret = readHeader(h->fd, &s);
	if (ret > 0){
		if (s.len > MAX_SCRIPT_LEN) {
			ASSERT(0);
		}
		ret = pipeRead(h->fd, (char *)&buff, s.len);
		if (ret > 0){
			ASSERT(ret == s.len);
			const char *argv[4] = {NULL};
			argv[0] = "sh";
			argv[1] = "-c";
			argv[2] = buff;
			argv[3] = NULL;
			switch(s.type) {
				case e_SYNC:
					pid = forkRunScript(argv, &s);
					DEBUG("shellsync(%05d)(%s)\n", pid, buff);
					p.pid = pid;
					pid_map[s.ev] = p;
					break;
				case e_ASYNC:
					pid = forkRunScript(argv, &s);
					DEBUG("shell    (%05d)(%s)\n", pid, buff);
					break;
				case e_KILL:
					ite = pid_map.find(s.ev);
					if(ite != pid_map.end()){
						p = pid_map[s.ev];
						int retval = kill(p.pid , SIGKILL);
						if (retval) {
							ERROR("shellover faild(%05d)\n", p.pid);
						}
					}
					break;
				default:
					ERROR("shell type error %d\n", s.type);
					break;
			}
		}
	}

	return;
}

static int childMainProcess()
{

#ifdef U_LOOP 
	uloop_init();
	child_handle[0].fd = parent_pipe[0];
	child_handle[0].cb = readScript;
	uloop_fd_add(&child_handle[0], ULOOP_READ);
	uloop_run();
	uloop_done();
#else
	signal(SIGCHLD, signalHandler);
	child_loop = ev_loop_new(EVBACKEND_EPOLL);
	ev_io_init (&child_handle[0], readScript, parent_pipe[0], EV_READ);
	ev_io_start (child_loop, &child_handle[0]);
	
	ev_run(child_loop);
	ev_loop_destroy(child_loop);
#endif

	ERROR("out child loop\n");	

	return -1;
}

int createChildProcess()
{
	pid_t pid;
	u32 buffersize = SOCKET_BUFFER_SIZE;
	
	if(socketpair(AF_UNIX, SOCK_STREAM, 0, parent_pipe) == -1) {
		ERROR("socketpair: %s\n", strerror(errno));
		return 1;
	}
	
	fcntl(parent_pipe[0], F_SETFL, O_NONBLOCK);
	fcntl(parent_pipe[1], F_SETFL, O_NONBLOCK);

	pid = fork();		
	switch (pid) {
		case -1:
			ERROR("fork %s\n",  strerror(errno));
			return 1;
		case 0:
			/* child  */ 	
			close(parent_pipe[1]);
			setsockopt(parent_pipe[0], SOL_SOCKET, SO_SNDBUF, 
						&buffersize, sizeof(buffersize));
			setsockopt(parent_pipe[0], SOL_SOCKET, SO_RCVBUF, 
						&buffersize, sizeof(buffersize));
			childMainProcess();
			exit(-1);
		default:
			/* parent */
			INFO("child pid =  %d\n", pid);
			close(parent_pipe[0]);
			setsockopt(parent_pipe[1], SOL_SOCKET, SO_SNDBUF, 
						&buffersize, sizeof(buffersize));
			setsockopt(parent_pipe[1], SOL_SOCKET, SO_RCVBUF, 
						&buffersize, sizeof(buffersize));
			break;
	}
	
	return 0;
}

